package cn.doitedu.rtmk.demo3;

import cn.doitedu.rtmk.demo1.EventBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;

/**
 * 相较 demo2的变化： 支持动态实时画像统计
 * 规则 1： 当 画像标签 age>30 and age<40 AND gender=male[静态画像条件]，在规则上线后，如果该用户发生了 3次+ w行为后 [实时统计画像标签条件] ，再发生 u 行为[触发条件]，立刻推出消息
 * 规则 2： 当 画像标签 active_level=3  AND gender=female[静态画像条件] 在规则上线后，用户依次发生过：(k, b ,c) [实时统计画像标签条件]，当发生 m行为(p1=v1)[触发条件]，推送消息
 */
public class Demo3 {
    public static void main(String[] args) throws Exception {

        // 创建编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 开启checkpoint
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.getCheckpointConfig().setCheckpointTimeout(2000);
        // 设置状态的backend
        env.setStateBackend(new HashMapStateBackend());


        // 构建 kafka source，读取用户实时行为数据
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                .setGroupId("doit40-1")
                .setTopics("dwd_events")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        // 读取kafka中的数据为一个流
        DataStreamSource<String> eventsStr = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s");

        // 解析行为日志为javabean
        SingleOutputStreamOperator<cn.doitedu.rtmk.demo1.EventBean> beanStream = eventsStr.map(json -> JSON.parseObject(json, cn.doitedu.rtmk.demo1.EventBean.class));


        // 把相同用户的行为，发到相同的subtask去处理
        KeyedStream<cn.doitedu.rtmk.demo1.EventBean, Long> keyedStream = beanStream.keyBy(EventBean::getUid);


        //
        keyedStream.process(new KeyedProcessFunction<Long, EventBean, String>() {

            Table userProfileTable;
            JSONObject jsonObject;


            ValueState<Integer> rule1State;
            ValueState<Integer> rule2State;
            String[] rule2ActionSequence;

            @Override
            public void open(Configuration parameters) throws Exception {

                org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
                conf.set("hbase.zookeeper.quorum", "doitedu:2181");

                Connection hbaseConnection = ConnectionFactory.createConnection(conf);
                userProfileTable = hbaseConnection.getTable(TableName.valueOf("user_profile"));

                jsonObject = new JSONObject();


                // 申请一个keyedState，来记录每个用户的规则1的w行为次数
                rule1State = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("rule-1-state", Integer.class));

                // 申请一个keyedState，来记录rule-2中的用户行为序列条件中，用户完成到了第几步
                rule2State = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("rule-2-state", Integer.class));

                rule2ActionSequence = new String[]{"k", "b", "c"};

            }

            @Override
            public void processElement(EventBean eventBean, KeyedProcessFunction<Long, EventBean, String>.Context context, Collector<String> collector) throws Exception {

                String eventId = eventBean.getEventId();
                /**
                 * 规则1
                 */
                // 判断事件是否是w，如果是，则做动态画像统计
                if (eventId.equals("w")) {
                    rule1State.update(rule1State.value() == null ? 0 : rule1State.value() + 1);
                }

                // 判断事件是否是u，如果是，则要判断该用户的所有规则条件是否已满足
                if (eventId.equals("u")) {


                    // 先判断该用户的 “动态画像条件”是否已满足
                    if (rule1State.value() != null && rule1State.value() >= 3) {

                        // 再去hbase查询静态画像条件是否满足
                        Get getParam = new Get(Bytes.toBytes(eventBean.getUid()));
                        getParam.addColumn(Bytes.toBytes("f"), Bytes.toBytes("age"));
                        getParam.addColumn(Bytes.toBytes("f"), Bytes.toBytes("gender"));

                        // 调用客户端查询
                        Result result = userProfileTable.get(getParam);
                        byte[] ageBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("age"));
                        byte[] genderBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("gender"));
                        int age = Integer.parseInt(Bytes.toString(ageBytes));
                        String genderStr = Bytes.toString(genderBytes);


                        // 判断该用户的画像标签值，是否满足规则要求的值
                        if (age >= 30 && age <= 40 && "male".equals(genderStr)) {

                            jsonObject.put("uid", eventBean.getUid());
                            jsonObject.put("timestamp", eventBean.getTimestamp());
                            jsonObject.put("rule_id", "rule-001");

                            collector.collect(jsonObject.toJSONString());

                        }
                    }

                }


                /**
                 * 规则2
                 */
                int currentStep = rule2State.value() == null ? 0 : rule2State.value();
                if (currentStep < 3 && eventId.equals(rule2ActionSequence[currentStep])) {
                    rule2State.update(currentStep + 1);
                }

                // 判断事件 bean中的 eventId=要求的触发事件
                if (eventBean.getEventId().equals("m") && eventBean.getProperties().getOrDefault("p1", "").equals("v1")) {

                    // 判断动态画像条件是否已满足
                    if (rule2State.value() != null && rule2State.value() >= 3) {

                        // 指定查询条件的行健
                        Get getParam = new Get(Bytes.toBytes(eventBean.getUid()));
                        // 指定要查询的标签（qualifier）
                        getParam.addColumn(Bytes.toBytes("f"), Bytes.toBytes("active_level"));
                        getParam.addColumn(Bytes.toBytes("f"), Bytes.toBytes("gender"));


                        // 调用客户端查询
                        Result result = userProfileTable.get(getParam);
                        byte[] activeLevelBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("active_level"));
                        byte[] genderBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("gender"));
                        int activeLevel = Integer.parseInt(Bytes.toString(activeLevelBytes));
                        String genderStr = Bytes.toString(genderBytes);

                        // 判断该用户的画像标签值，是否满足规则要求的值
                        if (activeLevel == 3 && "female".equals(genderStr)) {

                            jsonObject.put("uid", eventBean.getUid());
                            jsonObject.put("timestamp", eventBean.getTimestamp());
                            jsonObject.put("rule_id", "rule-002");

                            collector.collect(jsonObject.toJSONString());

                        }

                    }
                }


            }
        }).print();

        env.execute();

    }
}
